package com.atguigu.transform;

import com.atguigu.pojo.Event;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.file.src.FileSource;
import org.apache.flink.connector.file.src.reader.TextLineInputFormat;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.datastream.DataStreamSink;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.nio.charset.StandardCharsets;

/**
 * @Author Mr.Zheng
 * @Date 2023/6/14 16:16
 *
 * 富函数：相较于普通函数，富函数拥有更强大的功能。
 *      RichiFunction
 *          AbstractRichFunction
 *              RichMopFunction
 *              RichFilterFunction
 *              RichFlatmapFunction
 *              RichReduceFunction
 *              ......
 *
 * 富函数的功能：
 *      1. 拥有算子生命周期方法
 *              open（）：算子被创建时，执行一次
 *              close（）： 算子被销毁时，执行一次
 *      2. 获取到运行时上下文 RuntimeContext
 *              可以进行状态编程
 *                  getState（）
 *                  getListState()
 *                  getMapState()
 *                  getReducingState()
 *                  getAggregatingState()
 *
 *           获取更多其他的信息。
 */
public class Flink05_RichFunction {
    public static void main(String[] args) throws Exception {
        // 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 设置并行度
        env.setParallelism(1);

        // 读取socket文本数据，将数据封装成Event.打印输出

        //DataStreamSource<String> ds = env.socketTextStream("hadoop102", 8888);

        FileSource<String> fileSource = FileSource.forRecordStreamFormat(
                new TextLineInputFormat(StandardCharsets.UTF_8.name())
                ,
                new Path("input/clicks.txt")

        ).build();
        DataStreamSource<String> ds = env.fromSource(fileSource, WatermarkStrategy.noWatermarks(), "fileSource");

        ds.map(
                new RichMapFunction<String, Event>() {
                    /**
                     * 算子在创建时执行一次
                     */
                    @Override
                    public void open(Configuration parameters) throws Exception {
                        System.out.println("Flink05_RichFUnciton.open");
                    }

                    @Override
                    public Event map(String line) throws Exception {
                        System.out.println("Flink05_RichFunction.map");

                        //获取更多信息
                        //getRuntimeContext().getXXXXX

                        String[] fields = line.split(",");
                        return new Event(fields[0].trim(), fields[1].trim(), Long.valueOf(fields[2].trim()));

                    }

                    /**
                     * 算子在结束的时候执行一次
                     */
                    @Override
                    public void close() throws Exception {
                        System.out.println("Flink05_RichFunction.clost");
                    }
                }
        ).print();

         env.execute();
    }
}
